package demo.mqserver.datacenter;

import demo.common.MqException;
import demo.mqserver.core.Binding;
import demo.mqserver.core.Exchange;
import demo.mqserver.core.MSGQueue;
import demo.mqserver.core.Message;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;


//统一管理内存中的所有数据
//该类后续提供的一些方法，可能会在多线程环境下被使用
@Slf4j
public class MemoryDataCenter {

    // key 是 exchangeName    value 是 exchange 对象
    private ConcurrentHashMap<String, Exchange> exchangeMap = new ConcurrentHashMap<>();
    // key 是 queueName    value 是 MSGQueue 对象
    private ConcurrentHashMap<String, MSGQueue> queueMap = new ConcurrentHashMap<>();
    // 第一个 key 是 exchangeName    第二个 key 是 queueName
    private ConcurrentHashMap<String, ConcurrentHashMap<String, Binding>> bindingsMap = new ConcurrentHashMap<>();
    // key 是 messageId    value 是 Message 对象
    private ConcurrentHashMap<String, Message> messageMap = new ConcurrentHashMap<>();
    // key 是 queueName    value 是一个 Message 的链表
    private ConcurrentHashMap<String, LinkedList<Message>> queueMessageMap = new ConcurrentHashMap<>();
    //第一个 key 是 queueName    第二个 key 是 messageId
    private ConcurrentHashMap<String, ConcurrentHashMap<String, Message>> queueMessageWaitAckMap = new ConcurrentHashMap<>();

    // exchange
    public void insertExchange(Exchange exchange){
        exchangeMap.put(exchange.getName(), exchange);
        log.info("[MemoryDataCenter]新建交换机添加成功！exchangeName = " + exchange.getName());
    }

    public Exchange getExchange(String exchangeName){
        return exchangeMap.get(exchangeName);
    }

    public void deleteExchange(String exchangeName){
        exchangeMap.remove(exchangeName);
        log.info("[MemoryDataCenter]交换机删除成功！exchangeName = " + exchangeName);
    }


    // queue
    public void insertQueue(MSGQueue queue){
        queueMap.put(queue.getName(), queue);
        log.info("[MemoryDataCenter]新队列添加成功！queueName = " + queue.getName());
    }

    public MSGQueue getQueue(String queueName){
        return queueMap.get(queueName);
    }

    public void deleteQueue(String queueName){
        queueMap.remove(queueName);
        log.info("[MemoryDataCenter]队列删除成功！queueName = " + queueName);
    }


    // binding
    public void insertBinding(Binding binding) throws MqException {
        // 使用 exchangeName 检查对应的哈希表是否存在，不存在就创建一个
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.computeIfAbsent(binding.getExchangeName(),
                k -> new ConcurrentHashMap<>());
        //根据 queueName 检查，如果已经存在就抛出异常，不存在才能插入
        synchronized (bindingMap){
            if (bindingMap.get(binding.getQueueName()) != null) {
                throw new MqException("[MemoryDataCenter] 绑定已经存在！ exchangeName = " + binding.getExchangeName() + "，queueName = " + binding.getQueueName());
            }
            bindingMap.put(binding.getQueueName(), binding);
        }
        log.info("[MemoryDataCenter]新绑定添加成功！exchangeName = " + binding.getExchangeName() + "，queueName = " + binding.getQueueName());
    }

    //根据 exchangeName 和 queueName 获取唯一 binding
    public Binding getBinding(String exchangeName, String queueName){
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.get(exchangeName);
        if(bindingMap == null){
            return null;
        }
        return bindingMap.get(queueName);
    }

    //根据 exchangeName 获取所有的 binding
    public ConcurrentHashMap<String, Binding> getBindings(String exchangeName){
        return bindingsMap.get(exchangeName);
    }

    public void deleteBinding(Binding binding) throws MqException {
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.get(binding.getExchangeName());
        if(bindingMap == null){
            //该交换机没有绑定任何队列报错
            throw new MqException("[MemoryDataCenter]绑定不存在！exchangeName = " + binding.getExchangeName() + "，queueName = " + binding.getQueueName());
        }
        bindingMap.remove(binding.getQueueName());
        log.info("[MemoryDataCenter]绑定删除成功！exchangeName = " + binding.getExchangeName() + "，queueName = " + binding.getQueueName());
    }


    //message
    public void addMessage(Message message){
        messageMap.put(message.getMessageId(), message);
        log.info("[MemoryDataCenter]新消息添加成功！messageId = " + message.getMessageId());
    }

    public Message getMessage(String messageId){
        return messageMap.get(messageId);
    }

    public void removeMessage(String messageId){
        messageMap.remove(messageId);
        log.info("[MemoryDataCenter]消息删除成功！messageId = " + messageId);
    }

    //发送消息到指定队列
     public void sendMessage(MSGQueue queue, Message message){
        //把消息放到对应的队列数据结构中
        //根据队列的名字，找到该队列对应的消息队列
        LinkedList<Message> messages = queueMessageMap.computeIfAbsent(queue.getName(), k -> new LinkedList<>());
        //把数据加到 messages 里面
        synchronized (message){
            messages.add(message);
        }
        //把消息往消息中心插入一下，重复插入也没有关系
        addMessage(message);
        log.info("[MemoryDataCenter]消息被投递到队列中！message = " + message.getMessageId());
     }

    //从队列中取消息
    public Message pollMessage(String queueName){
        //根据队列名，查找对应的队列的消息链表
        LinkedList<Message> messages = queueMessageMap.get(queueName);
        if(messages == null){
            return null;
        }
        synchronized (messages){
            //队列中没有消息
            if(messages.size() == 0){
                return null;
            }
            //链表中有元素，进行头删
            Message currentMessage = messages.remove(0);
            log.info("[MemoryDataCenter]消息从队列中去除！messageId = " + currentMessage.getMessageId());
            return currentMessage;
        }
    }

    //获取指定队列中消息的个数
    public int getMessageCount(String queueName){
        LinkedList<Message> messages = queueMessageMap.get(queueName);
        if(messages == null){
            //队列中没有消息
            return 0;
        }
        synchronized (messages){
            return messages.size();
        }
    }

    //添加未确认消息
    public void addMessageWaitAck(String queueName, Message message){
        ConcurrentHashMap<String, Message> messageHashMap = queueMessageWaitAckMap.computeIfAbsent(queueName, k -> new ConcurrentHashMap<>());
        messageHashMap.put(message.getMessageId(), message);
        log.info("[MemoryDataCenter]消息进入待确认队列！messageId = " + message.getMessageId());
    }

    //删除未确认消息（消息已经确认过了）
    public void removeMessageWaitAck(String queueName, String messageId){
        ConcurrentHashMap<String, Message> messageHashMap = queueMessageWaitAckMap.get(queueName);
        if(messageHashMap == null){
            return ;
        }
        messageHashMap.remove(messageId);
        log.info("[MemoryDataCenter]消息从待确认队列删除！messageId = " + messageId);
    }

    //获取未确认消息
    public Message getMessageWaitAck(String queueName, String messageId){
        ConcurrentHashMap<String, Message> messageHashMap = queueMessageWaitAckMap.get(queueName);
        if(messageHashMap == null){
            return null;
        }
        return messageHashMap.get(messageId);
    }


    //从硬盘上读取数据，把硬盘上之前持久化存储的各个维度的数据都恢复到内存中
    public void recovery(DiskDataCenter diskDataCenter) throws IOException, MqException, ClassNotFoundException {
        //0.清空之前的所有数据
        exchangeMap.clear();
        queueMap.clear();
        bindingsMap.clear();
        messageMap.clear();
        queueMessageMap.clear();

        //1.恢复所有的交换机数据
        List<Exchange> exchanges = diskDataCenter.selectAllExchanges();
        for(Exchange exchange : exchanges){
            exchangeMap.put(exchange.getName(), exchange);
        }

        //2.恢复所有的队列数据
        List<MSGQueue> queues = diskDataCenter.selectAllQueues();
        for(MSGQueue queue : queues){
            queueMap.put(queue.getName(), queue);
        }

        //3.恢复所有的绑定数据
        List<Binding> bindings = diskDataCenter.selectAllBindings();
        for(Binding binding : bindings){
            ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.computeIfAbsent(binding.getExchangeName(), k -> new ConcurrentHashMap<>());
            bindingMap.put(binding.getQueueName(), binding);
        }

        //4.恢复所有的消息数据
        //遍历所有的队列，根据每个队列的名字，获取到所有的消息
        for(MSGQueue queue : queues){
            LinkedList<Message> messages = diskDataCenter.loadAllMessageFromQueue(queue.getName());
            queueMessageMap.put(queue.getName(), messages);
            for(Message message : messages){
                messageMap.put(message.getMessageId(), message);
            }
        }

        //针对“未确认的消息”这部分内存中的数据，不需要从硬盘中恢复，在考虑硬盘存储时也没设定这一块
        //在等待 ack 过程中，服务器重启了，此时“未被确认的数据”就恢复成“未取走的数据”
        //在硬盘上存储时，就当做“未取走”
    }
}
